package com.passion.source;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.apache.flume.Context;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.SimpleEvent;
import org.apache.flume.source.AbstractSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.Serializable;
import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @program flume-component
 * @Description 采集数据库变更数据
 * @Author limeihua
 * @Date 2024/09/02 15:26
 * @Version 1.0
 */
public class MysqlCDCSource extends AbstractSource implements Configurable, EventDrivenSource {
    private String host;
    private int port;
    private String username;
    private String password;
    private String database;
    private BinaryLogClient client;
    private String tableName;
    private Map<Long, String[]> tableColumns = new HashMap<>();
    private static String monitorTable;
    private static final Logger logger = LoggerFactory.getLogger(MysqlCDCSource.class);
    @Override
    public void configure(Context context) {
        host = context.getString("mysqlHost", "192.168.46.30");
        port = context.getInteger("mysqlPort", 3306);
        username = context.getString("mysqlUsername", "root");
        password = context.getString("mysqlPassword", "159357");
        database = context.getString("mysqlDatabase");
        monitorTable = context.getString("monitorTable");
    }

    @Override
    public synchronized void start() {
        client = new BinaryLogClient(host, 3306, username, password);
        String targetDatabase = database;
        logger.info("cdc连接为：" + client);
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
        client.registerEventListener(event -> {
            EventData data = event.getData();
            if (data instanceof TableMapEventData) {
                TableMapEventData tableMapEventData = (TableMapEventData) data;
                // 检查表所属的数据库是否与目标数据库匹配
                if (tableMapEventData.getDatabase().equals(targetDatabase)) {
                    // 获取表的元数据信息并存储
                    tableName = tableMapEventData.getTable();
                    Long tableId = tableMapEventData.getTableId();
                    String[] columns = fetchTableColumns(tableName);
                    tableColumns.put(tableId, columns);
                }
            } else if (data instanceof WriteRowsEventData) {
                WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
                Long tableId = writeRowsEventData.getTableId();
                String[] columns = tableColumns.get(tableId);
                if (columns != null) {
                    for (Serializable[] row : writeRowsEventData.getRows()) {
                        if(monitorTable.equals(tableName)){
                            String value = "";
                            for (int i = 0; i < row.length; i++) {
                                if(i == row.length - 1){
                                    value = value + row[i];
                                }else {
                                    value = value + row[i] + "\t";
                                }
                            }
                            SimpleEvent flumeEvent = new SimpleEvent();
                            flumeEvent.setBody(value.getBytes());
                            getChannelProcessor().processEvent(flumeEvent);
                            logger.info("已发送到Channel的数据为：" + value);
                        }
                    }
                }
            } else if (data instanceof UpdateRowsEventData) {
                UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
                Long tableId = updateRowsEventData.getTableId();
                String[] columns = tableColumns.get(tableId);
                if (columns != null) {
                    for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
                        if(monitorTable.equals(tableName)){
                            Serializable[] key = row.getKey();
                            Serializable[] value = row.getValue();
                            String valueString = "";
                            for (int i = 0; i < key.length; i++) {
                                if(i == key.length - 1){
                                    valueString = valueString + value[i];
                                }else {
                                    valueString = valueString + value[i] + "\t";
                                }
                            }
                            SimpleEvent flumeEvent = new SimpleEvent();
                            flumeEvent.setBody(valueString.getBytes());
                            getChannelProcessor().processEvent(flumeEvent);
                            logger.info("已发送到Channel的数据为：" + valueString);
                        }
                    }
                }
            } else if (data instanceof DeleteRowsEventData) {
                DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
                Long tableId = deleteRowsEventData.getTableId();
                String[] columns = tableColumns.get(tableId);
                if (columns != null) {
                    for (Serializable[] row : deleteRowsEventData.getRows()) {
//                        JSONObject rowData = new JSONObject();
//                        for (int i = 0; i < row.length; i++) {
//                            rowData.put(columns[i], row[i]);
//                        }
//                        JSONObject output = new JSONObject();
//                        SimpleEvent flumeEvent = new SimpleEvent();
//                        flumeEvent.setBody(output.toString().getBytes());
//                        getChannelProcessor().processEvent(flumeEvent);
//                        logger.info("已发送到Channel的数据为：" + output.toString());
                        if(monitorTable.equals(tableName)){
                            String value = "";
                            for (int i = 0; i < row.length; i++) {
                                if(i == row.length - 1){
                                    value = value + row[i];
                                }else {
                                    value = value + row[i] + "\t";
                                }
                            }
                            SimpleEvent flumeEvent = new SimpleEvent();
                            flumeEvent.setBody(value.getBytes());
                            getChannelProcessor().processEvent(flumeEvent);
                            logger.info("已发送到Channel的数据为：" + value);
                        }
                    }
                }
            }
        });
        try {
            client.connect();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public synchronized void stop() {
        try {
            client.disconnect();
        } catch (IOException e) {
            throw new RuntimeException("Failed to disconnect from MySQL server", e);
        }
    }
    private String[] fetchTableColumns(String tableName) {
        List<String> columns = new ArrayList<>();
        try (Connection conn = DriverManager.getConnection("jdbc:mysql://"+ host+ ":3306/" + database, username, password)) {
            DatabaseMetaData metaData = conn.getMetaData();
            try (ResultSet rs = metaData.getColumns(null, null, tableName, null)) {
                while (rs.next()) {
                    String columnName = rs.getString("COLUMN_NAME");
                    columns.add(columnName);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return columns.toArray(new String[0]);
    }
}

